53. Pandas模块的统计计算

# 计算各项统计量
stats_overview = df_returns.agg(['mean', 'std', 'min', 'max'])  # 将四项核心统计量汇总为紧凑对照表
print('核心统计量对照:')  # 标明可见输出对应均值、波动与取值边界
print(stats_overview.round(4))  # 输出四项模拟资产的任务特异统计结果

# 四分位距(IQR)
Q1 = df_returns.quantile(0.25)
Q3 = df_returns.quantile(0.75)
IQR = Q3 - Q1
print('四分位距(IQR):')
print(IQR)

# 识别异常值(超出1.5*IQR)
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 检测异常值
outliers = (df_returns < lower_bound) | (df_returns > upper_bound)
print(f'\n异常值数量:')
print(outliers.sum())

skewness = df_returns.skew()
kurtosis = df_returns.kurtosis()

distribution_check = []  # 准备用于课堂核验的分布形状摘要

# 解释
for stock in df_returns.columns:
    skew_val = skewness[stock]
    kurt_val = kurtosis[stock]
    if skew_val > 0.5:
        skew_interp = '右偏,极端正收益更多'
    elif skew_val < -0.5:
        skew_interp = '左偏,极端负收益更多'
    else:
        skew_interp = '近似对称'
    if kurt_val > 1:
        kurt_interp = '尖峰厚尾'
    elif kurt_val < -1:
        kurt_interp = '低峰薄尾'
    else:
        kurt_interp = '接近正态'
    distribution_check.append({'模拟资产': stock, '偏度': skew_val, '峰度': kurt_val, '偏度判断': skew_interp, '峰度判断': kurt_interp})  # 保存每项模拟资产的数值与解释
distribution_summary = pd.DataFrame(distribution_check).set_index('模拟资产')  # 合并为四行便于核对结果表
print(distribution_summary.round({'偏度': 3, '峰度': 3}))  # 输出紧凑的偏度、峰度及解释

prices = pd.DataFrame({  # 构造两项匿名模拟资产的价格路径
    '模拟资产A': [100, 102, 101, 104, 103, 106],  # 设置模拟资产A的课堂价格序列
    '模拟资产B': [100, 99, 101, 103, 102, 105]  # 设置模拟资产B的课堂价格序列
})

price_changes = prices.diff()  # 计算可加的逐期价格变化而非直接累加价格水平
cumulative_changes = price_changes.fillna(0).cumsum()  # 累加逐期变化以重构相对期初的价格变动
period_returns = prices.pct_change().fillna(0)  # 计算逐期收益率并将首期基准设为零
wealth_index = (1 + period_returns).cumprod()  # 用复利规则生成以1为起点的财富指数
cumulative_check = pd.DataFrame({'期初价格': prices.iloc[0], '期末价格': prices.iloc[-1], '累计价格变动': cumulative_changes.iloc[-1], '复合财富指数': wealth_index.iloc[-1]})  # 汇总变化累加与收益复利结果
assert np.allclose(cumulative_changes.iloc[-1], prices.iloc[-1] - prices.iloc[0])  # 验证变化累加等于期末减期初
assert np.allclose(wealth_index.iloc[-1], prices.iloc[-1] / prices.iloc[0])  # 验证复合财富等于价格相对期初的比率
print('变化累加与收益复利:')  # 标明两种累积对象不可混用
print(cumulative_check.round(4))  # 输出两项模拟资产的便于核对结果

# 累计收益率
initial_prices = prices.iloc[0]
cum_returns = (prices / initial_prices - 1) * 100
cumulative_extrema = pd.DataFrame({'历史最高': prices.cummax().iloc[-1], '历史最低': prices.cummin().iloc[-1], '累计收益率(%)': cum_returns.iloc[-1]})  # 汇总期末便于核对的累积极值与收益率
print('价格极值与期末累计收益率:')  # 标明输出用于比较两项模拟资产
print(cumulative_extrema.round(2))  # 输出紧凑的累计统计结果

np.random.seed(42)  # 固定随机种子使每次课堂模拟结果一致
simulated_returns = pd.Series(np.random.normal(0.001, 0.02, 100), index=pd.date_range('2024-01-01', periods=100), name='模拟日收益率')  # 生成匿名课堂模拟收益率
compounded_wealth = (1 + simulated_returns).cumprod()  # 按逐期收益率复利生成财富指数
nav = pd.DataFrame({'日期': compounded_wealth.index, '净值': compounded_wealth.to_numpy()})  # 组装复合净值时间序列

# 计算历史最高点与回撤
nav['历史最高'] = nav['净值'].cummax()
nav['回撤'] = (nav['净值'] - nav['历史最高']) / nav['历史最高']

max_drawdown = nav['回撤'].min()
drawdown_check = nav.nsmallest(3, '回撤')[['日期', '净值', '历史最高', '回撤']]  # 选取回撤最深的三个观测用于核验
assert (nav['净值'] > 0).all()  # 验证复合财富路径始终为正
assert nav['回撤'].between(-1, 0).all()  # 验证回撤位于经济上有效的负一到零区间
assert np.isclose(max_drawdown, nav['回撤'].min())  # 验证最大回撤等于逐期回撤最小值
print(f'最大回撤: {max_drawdown:.2%}')  # 输出核心风险指标
print(drawdown_check)  # 输出最大回撤附近的任务特异依据

dates = pd.date_range('2024-01-01', periods=100)
prices_ts = pd.DataFrame({
    '日期': dates,
    '收盘价': 100 + np.cumsum(np.random.normal(0.5, 2, 100))
})
prices_ts = prices_ts.set_index('日期')

# 计算滚动统计量
prices_ts['MA5'] = prices_ts['收盘价'].rolling(window=5).mean()
prices_ts['MA20'] = prices_ts['收盘价'].rolling(window=20).mean()
prices_ts['STD5'] = prices_ts['收盘价'].rolling(window=5).std()

print('滚动统计量(最后10行):')
print(prices_ts.tail(10))

prices_ts['布林带_上'] = prices_ts['MA20'] + 2 * prices_ts['收盘价'].rolling(window=20).std()
prices_ts['布林带_下'] = prices_ts['MA20'] - 2 * prices_ts['收盘价'].rolling(window=20).std()

bollinger_check = prices_ts[['收盘价', 'MA20', '布林带_上', '布林带_下']].tail(2)  # 选取末两期形成布林带核验摘要
print('布林带末两期:')  # 标明摘要保留最新技术指标状态
print(bollinger_check.round(4))  # 输出收盘价与上下轨的任务特异结果

prices_ts['累积均值'] = prices_ts['收盘价'].expanding().mean()
prices_ts['累积标准差'] = prices_ts['收盘价'].expanding().std()
prices_ts['累积最大值'] = prices_ts['收盘价'].expanding().max()

# 年化累计波动率
prices_ts['累计波动率'] = prices_ts['收盘价'].pct_change().expanding().std() * np.sqrt(252)  # 计算随样本扩展更新的年化波动率
expanding_check = prices_ts[['收盘价', '累积均值', '累积标准差', '累积最大值', '累计波动率']].tail(3)  # 保留末三期形成便于核对摘要
print('扩展窗口末三期统计:')  # 标明摘要的时间范围
print(expanding_check.round(4))  # 输出长期统计量的最新水平与变化

industry_stats = industry_data.groupby('行业').agg({
    '市盈率': ['mean', 'median', 'std'],
    '市净率': ['mean', 'min', 'max'],
    'ROE': 'mean',
    '股息率': 'mean'
})
print('行业详细统计:')
print(industry_stats.round(4))

# 分组计数
print('\n行业股票数量:')
print(industry_data.groupby('行业').size())

multi_level_data = pd.DataFrame({
    '行业': ['白酒', '白酒', '白酒', '银行', '银行', '银行', '保险', '保险'],
    '市值等级': ['大盘', '中盘', '大盘', '大盘', '小盘', '大盘', '大盘', '中盘'],
    '市盈率': [45.2, 35.8, 38.5, 8.5, 15.2, 6.8, 12.3, 18.5],
    '收益率': [0.05, 0.03, 0.06, 0.02, 0.04, 0.01, 0.03, 0.02]
})

df_multi = pd.DataFrame(multi_level_data)

# 多级分组
multi_stats = df_multi.groupby(['行业', '市值等级']).agg({
    '市盈率': 'mean',
    '收益率': 'mean'
})
print('多级分组统计:')
print(multi_stats)

# 自定义：基于ROE计算课堂标准化分数，不称为夏普比率
def roe_classroom_standardization(series, benchmark_roe=0.10):  # 定义课堂ROE标准化函数
    '''计算ROE均值相对课堂基准的组内标准化幅度'''  # 说明该量不是收益风险比率
    group_std = series.std()  # 计算组内ROE样本标准差
    return (series.mean() - benchmark_roe) / group_std if group_std > 0 else 0  # 标准化均值与课堂基准的距离

industry_groups = industry_data.groupby('行业')  # 建立课堂行业分组对象
pb_pe_mean = (industry_data['市净率'] / industry_data['市盈率']).groupby(industry_data['行业']).mean()  # 计算各行业市净率市盈率比率均值
roe_classroom_score = industry_groups['ROE'].apply(roe_classroom_standardization)  # 计算ROE课堂标准化分数
pe_range = industry_groups['市盈率'].agg(lambda values: values.max() - values.min())  # 计算各行业市盈率范围
custom_summary = pd.DataFrame({'市净率/市盈率均值': pb_pe_mean, 'ROE课堂标准化分数': roe_classroom_score, '市盈率范围': pe_range})  # 合并三项课堂聚合结果
print(custom_summary.round(4))  # 输出匿名课堂行业的自定义统计摘要

# 使用之前的课堂模拟收益率数据
cov_matrix = df_returns.cov()
print('协方差矩阵:')
print(cov_matrix.round(6))

corr_matrix = df_returns.corr()
print('\n相关系数矩阵:')
print(corr_matrix.round(4))

upper_mask = np.triu(np.ones(corr_matrix.shape, dtype=bool), k=1)  # 仅保留相关矩阵上三角以排除自相关和镜像重复
unique_pairs = corr_matrix.where(upper_mask).stack().rename('相关系数').reset_index()  # 展开唯一跨资产相关对且不按数值删除完全相关
unique_pairs.columns = ['资产1', '资产2', '相关系数']  # 设置便于业务阅读的字段名
unique_pairs['绝对相关系数'] = unique_pairs['相关系数'].abs()  # 计算排序所需的绝对相关强度
signed_top = unique_pairs.loc[unique_pairs['相关系数'].idxmax()]  # 用原值选择最大带符号相关的唯一资产对
absolute_top = unique_pairs.loc[unique_pairs['相关系数'].abs().idxmax()]  # 用绝对值选择方向不限的最强相关唯一资产对
expected_pair_count = corr_matrix.shape[0] * (corr_matrix.shape[0] - 1) // 2  # 计算无序跨资产对的理论数量
assert len(unique_pairs) == expected_pair_count  # 验证每个无序资产对只保留一次
assert not unique_pairs[['资产1', '资产2']].duplicated().any()  # 验证结果不存在镜像或重复资产对
perfect_probe = pd.DataFrame({'模拟资产甲': [1, 2, 3], '模拟资产乙': [2, 4, 6]}).corr()  # 构造跨资产完全相关的课堂边界探针
perfect_probe_mask = np.triu(np.ones(perfect_probe.shape, dtype=bool), k=1)  # 为边界探针保留唯一上三角资产对
perfect_probe_pairs = perfect_probe.where(perfect_probe_mask).stack()  # 提取跨资产完全相关值而不与对角线混淆
assert np.isclose(perfect_probe_pairs.iloc[0], 1.0)  # 验证算法保留真实跨资产完全相关系数
print('最大带符号相关唯一对:',signed_top[['资产1','资产2','相关系数']].to_dict())  # 输出原值 idxmax 的带符号口径
print('绝对值最强相关唯一对:',absolute_top[['资产1','资产2','相关系数']].to_dict())  # 输出绝对值口径并保留符号